package demo.delay;

import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;

import javax.annotation.Resource;
import java.util.Arrays;


public class KafkaDltMessageHandler {

    private Logger log = LoggerFactory.getLogger(KafkaDelayConsumerThread.class);

    @Resource
    private KafkaAdminClient kafkaAdminClient;

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Value("${spring.cloud.stream.kafka.num-partitions:3}")
    private int numPartitions;

    @Value("${spring.cloud.stream.kafka.replication-factor:1}")
    private int replicationFactor;

    public static final String DLT_TOPIC_SUFFIX = ".DLT";

    public void putMessage(String topic, String message) {
        if (!topic.endsWith(DLT_TOPIC_SUFFIX)) {
            topic += DLT_TOPIC_SUFFIX;
        }

        NewTopic newTopic = new NewTopic(topic, numPartitions, (short) replicationFactor);

        kafkaAdminClient.createTopics(Arrays.asList(newTopic));

        kafkaTemplate.send(topic, message.getBytes());

        log.info("KafkaDltMessageHandler: create topic: {}, success, put message: {}", topic, message);
    }


}